package com.theaf.core.rmq;

import com.rabbitmq.client.*;
import com.theaf.core.rmq.common.Constants;
import com.theaf.core.rmq.common.MQMessage;
import com.theaf.core.rmq.common.MqReturn;
import com.theaf.core.utils.ApplicationProperties;
import com.theaf.core.utils.SmExceptionHelper;
import com.theaf.core.utils.SpringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;

import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;

import static com.theaf.core.rmq.common.Constants.MES_REP_SIZE;

/**
 * Created by littlersmall on 16/5/11.
 */
@Slf4j
class MQAccessBuilder {
    private ConnectionFactory connectionFactory;

    private final String RPC_QUEUE = "rpcQueue_smyw_2018";

    public static String INVOKE_QUEUE = "invokeQueue_smyw_2018";

    /**
     * 存储各发送的回调类
     */
    private Map<String, RpcCallBack> rpcMap = new ConcurrentHashMap();

    public MQAccessBuilder(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }


    /**
     * 构造发送者
     * @return
     * @throws IOException
     */
    public MessageSender buildMessageSender() throws IOException {
        Connection connection = connectionFactory.createConnection();
        //创建用于回调的队列，每个系统一个
        String rpcQueue = RPC_QUEUE+ ApplicationProperties.getProperty("spring.dubbo.application.name");
        ThreadPoolConsumer.addConsumer(buildMessageConsumer(null, "", rpcQueue, new MessageProcess() {
            @Override
            public MqReturn process(Object message, int excuteNumber, String id) {
                if(message instanceof MQMessage){
                    MQMessage mes = (MQMessage)message;
                    RpcCallBack rpc = rpcMap.get(mes.getId());
                    if(rpc!=null){
                        rpc.callBack(mes.getBody());
                        rpc = null;
                        rpcMap.remove(mes.getId());
                    }
                }
                return null;
            }
        }));
        //创建用于本地服务调用的队列，每个系统一个
        INVOKE_QUEUE = INVOKE_QUEUE+ ApplicationProperties.getProperty("spring.dubbo.application.name");
        ThreadPoolConsumer.addConsumer(buildMessageConsumer(null, "", INVOKE_QUEUE, new MessageProcess() {
            @Override
            public MqReturn process(Object message,int excuteNumber,String id) {
                if(message instanceof MQMessage){
                    MQMessage mes = (MQMessage)message;
                    Object[] obs = (Object[])mes.getBody();
                    Object bean = SpringUtils.getBean((String)obs[0]);
                    for(java.lang.reflect.Method method :bean.getClass().getDeclaredMethods()){
                        if(method.getName().equals(obs[1])){
                            try {
                                method.invoke(bean,obs[2]);
                            } catch (Exception e) {
                                throw SmExceptionHelper.sysCheckException(e.getMessage());
                            }
                        }
                    }
                }
                return null;
            }
        }));
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());

        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                log.info("send message failed: " + cause + correlationData.toString());
            } else {

            }
        });
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, tmpExchange, tmpRoutingKey) -> {
            try {
                Thread.sleep(Constants.ONE_SECOND);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info("2send message failed: " + replyCode + " " + replyText);
            rabbitTemplate.send(tmpExchange,tmpRoutingKey,message);
        });
        return new MessageSender() {


            @Override
            public void send(Object message,String exchange,String routingKey,
                             RpcCallBack rpcCallBack,long deplay) throws Exception{
                String id = UUID.randomUUID().toString().replaceAll("-","");
                if(rpcCallBack!=null){
                    rpcMap.put(id,rpcCallBack);
                }
                rabbitTemplate.convertAndSend(exchange,routingKey,new MQMessage(id,message,
                                System.currentTimeMillis()/1000,deplay,rpcCallBack!=null,rpcQueue),
                        new CorrelationData(id));
            }
        };
    }

    public <T> MessageConsumer buildMessageConsumer(String exchange, String routingKey, final String queue,
                                                    final MessageProcess<T> messageProcess) throws IOException {
        return buildMessageConsumer(exchange, routingKey, queue, messageProcess, "direct");
    }

    public <T> MessageConsumer buildTopicMessageConsumer(String exchange, String routingKey, final String queue,
                                                         final MessageProcess<T> messageProcess) throws IOException {
        return buildMessageConsumer(exchange, routingKey, queue, messageProcess, "topic");
    }

    public <T> MessageConsumer buildFanoutMessageConsumer(String exchange, String routingKey, final String queue,
                                                         final MessageProcess<T> messageProcess) throws IOException {
        return buildMessageConsumer(exchange, routingKey, queue, messageProcess, "fanout");
    }

    //1 创建连接和channel
    //2 设置message序列化方法
    //3 构造consumer
    public <T> MessageConsumer buildMessageConsumer(String exchange, String routingKey, final String queue,
                                                    final MessageProcess<T> messageProcess, String type) throws IOException {
        final Connection connection = connectionFactory.createConnection();

        //1
        buildQueue(exchange, routingKey, queue, connection, type);

        //2
        final MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();
        final MessageConverter messageConverter = new Jackson2JsonMessageConverter();

        //3
        return new MessageConsumer() {
            QueueingConsumer consumer;

            {
                consumer = buildQueueConsumer(connection, queue);
            }

            /**
             * 消息重发次数映射
             */
            private Map<String, Integer> mesRepSizeMap = new ConcurrentHashMap();

            @Override
            //1 通过delivery获取原始数据
            //2 将原始数据转换为特定类型的包
            //3 处理数据
            //4 手动发送ack确认
            public void consume() {
                QueueingConsumer.Delivery delivery;
                Channel channel = consumer.getChannel();
                try {
                    delivery = consumer.nextDelivery();
                    Message message = new Message(delivery.getBody(),
                            messagePropertiesConverter.toMessageProperties(delivery.getProperties(), delivery.getEnvelope(), "UTF-8"));

                    MQMessage messageBean = (MQMessage) messageConverter.fromMessage(message);

                    MqReturn detailRes = null;
                    boolean success = false;
                    try {
                        long current = System.currentTimeMillis()/1000;
                        //延迟时间未到放回队列
                        if(messageBean.getSendTime()+messageBean.getDelay()>current) {
                            channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
                            return ;
                        }
                        Integer n = mesRepSizeMap.get(messageBean.getId());
                        detailRes = messageProcess.process((T) messageBean.getBody(),n==null?1:(n+1),messageBean.getId());
                        if(detailRes==null||detailRes.isSuccess()) {
                            success = true;
                            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                            if (messageBean.isRpced()) {
                                MQMessage mqMessage = new MQMessage(messageBean.getId(), detailRes, 0, 0, false, null);
                                channel.basicPublish("", messageBean.getRpcQueue(), null, messageConverter.toMessage(messageBean, new MessageProperties()).getBody());
                            }
                        }
                    } catch (Exception e) {
                        success = false;
                    }finally {
                        if(!success) {
                            //避免过多失败log
                            Thread.sleep(Constants.ONE_SECOND);
                            Integer mesRepSize = mesRepSizeMap.get(messageBean.getId());
                            if (mesRepSize == null) {
                                mesRepSize = 0;
                            }
                            if (mesRepSize++ < MES_REP_SIZE) {
                                channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
                                mesRepSizeMap.put(messageBean.getId(), mesRepSize);
                            } else {
                                mesRepSizeMap.remove(messageBean.getId());
                            }
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ShutdownSignalException | ConsumerCancelledException | IOException e) {
                    e.printStackTrace();

                    try {
                        channel.close();
                    } catch (IOException | TimeoutException ex) {
                        ex.printStackTrace();
                    }

                    consumer = buildQueueConsumer(connection, queue);
                } catch (Exception e) {
                    e.printStackTrace();
                    log.info("exception : ", e);

                    try {
                        channel.close();
                    } catch (IOException | TimeoutException ex) {
                        ex.printStackTrace();
                    }

                    consumer = buildQueueConsumer(connection, queue);
                }
            }
        };
    }

    private void buildQueue(String exchange, String routingKey,
                            final String queue, Connection connection, String type) throws IOException {
        Channel channel = connection.createChannel(false);

        if(exchange!=null) {
            if (type.equals("direct")) {
                channel.exchangeDeclare(exchange, "direct", true, false, null);
            } else if (type.equals("topic")) {
                channel.exchangeDeclare(exchange, "topic", true, false, null);
            } else {
                channel.exchangeDeclare(exchange, "fanout", true, false, null);
            }
        }

        channel.queueDeclare(queue, true, false, false, null);
        if(exchange!=null) {
            channel.queueBind(queue, exchange, routingKey);
        }

        try {
            channel.close();
        } catch (TimeoutException e) {
            e.printStackTrace();
            log.info("close channel time out ", e);
        }
    }

    private void buildTopic(String exchange, Connection connection) throws IOException {
        Channel channel = connection.createChannel(false);
        channel.exchangeDeclare(exchange, "topic", true, false, null);
    }

    private QueueingConsumer buildQueueConsumer(Connection connection, String queue) {
        try {
            Channel channel = connection.createChannel(false);
            QueueingConsumer consumer = new QueueingConsumer(channel);

            //通过 BasicQos 方法设置prefetchCount = 1。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。
            //换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它
            channel.basicQos(1);
            channel.basicConsume(queue, false, consumer);
            return consumer;
        } catch (Exception e) {
            e.printStackTrace();
            log.info("build queue consumer error : ", e);

            try {
                Thread.sleep(Constants.ONE_SECOND);
            } catch (InterruptedException inE) {
                inE.printStackTrace();
            }

            return buildQueueConsumer(connection, queue);
        }
    }

    //for test
    public int getMessageCount(final String queue) throws IOException {
        Connection connection = connectionFactory.createConnection();
        final Channel channel = connection.createChannel(false);
        final AMQP.Queue.DeclareOk declareOk = channel.queueDeclarePassive(queue);

        return declareOk.getMessageCount();
    }
}
